package sample.batch.scope;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sample.batch.helloworld.Person;
import sample.batch.helloworld.PersonItemProcessor;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;

/**
 * Created on 2020/8/4
 */
@SpringBootApplication
@EnableBatchProcessing
public class ScopeApplication {
	private static final Logger LOGGER = LoggerFactory.getLogger(ScopeApplication.class);


	public static void main(String[] args) throws Exception {
		System.exit(SpringApplication.exit(SpringApplication.run(ScopeApplication.class, args)));
	}

	@Autowired
	public JobBuilderFactory jobBuilderFactory;

	@Autowired
	public StepBuilderFactory stepBuilderFactory;

	@Bean
	public FlatFileItemReader<Person> reader() {
		return new FlatFileItemReaderBuilder<Person>()
				.name("personItemReader")
				.resource(new ClassPathResource("sample-data.csv"))
				.delimited()
				.names(new String[]{"firstName", "lastName"})
				.fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
					setTargetType(Person.class);
				}})
				.build();
	}

	@Bean
	public PersonItemProcessor processor() {
		return new PersonItemProcessor();
	}

	@Bean
	@StepScope
	public ItemWriter<Person> writer() {
		return items -> {
			for (Person person : items) {
				LOGGER.info(toString() + ":" + person.toString());
			}
		};
	}

	@Bean
	public Job scopeJob() {
		return jobBuilderFactory.get("scopeJob")
				.start(step1())
				.next(step2())
				//.preventRestart()
				.build();
	}

	@Bean
	public Step step1() {
		return stepBuilderFactory.get("step1")
				.<Person, Person> chunk(3)
				.reader(reader())
				.processor(processor())
				.writer(writer())
				.allowStartIfComplete(true)
				.taskExecutor(taskExecutor())
				.build();
	}

	@Bean
	public TaskExecutor taskExecutor() {
		SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
		taskExecutor.setConcurrencyLimit(3);
		return taskExecutor;
	}

	@Bean
	public Step step2() {
		return stepBuilderFactory.get("step2")
				.<Person, Person> chunk(3)
				.reader(reader())
				.processor(processor())
				.writer(writer())
				.allowStartIfComplete(true)
				.build();
	}
}
